package com.hubing.cloud.app.web.es;

import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.ResourceAlreadyExistsException;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.engine.DocumentMissingException;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.metrics.CardinalityAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedCardinality;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.collapse.CollapseBuilder;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * ES API (可以不走spring)
 */
@Component
public class ElasticSearchRestManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchRestManager.class);

    private static final int ITEMS_PER_SEARCH = 20;

    public static final int INDEX_PER_PAGE_SIZE = 50;

    /**
     * ES查询时最大的长度
     */
    private static final int ES_MAX_SIZE = 100000;

    private static RestHighLevelClient client = null;

    public static final String ES_INDEX = "index";
    public static final String ES_TYPE = "type";
    public static final String CONNECTION_TYPE = "http";

    private ElasticSearchRestManager() {
    }

    public static void initElasticSearch() {
        // 避免redis nio 与es nio冲突
        System.setProperty("es.set.netty.runtime.available.processors", "false");
        initTransportClient();
    }

    private static BulkProcessor.Listener listener = new BulkProcessor.Listener() {
        @Override
        public void beforeBulk(long executionId, BulkRequest request) {
            // 是否提交后里面刷新
            request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request,
                              BulkResponse response) {
            // 提交结束后调用（无论成功或失败）
            if (response.hasFailures()) {
                LOGGER.error("提交{}个文档，用时{}MS,有文档提交失败！失败原因：{}", response.getItems().length,
                        response.getTook().getMillis(), response.buildFailureMessage());
            } else {
                LOGGER.info("提交{}个文档，用时{}MS！", response.getItems().length, response.getTook().getMillis());
            }
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request,
                              Throwable failure) {
            LOGGER.error("Failed to execute bulk", failure);
        }
    };

    private static BulkProcessor bulkProcessor = BulkProcessor.builder(
                    (request, bulkListener) ->
                            ElasticSearchRestManager.getClient().bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
                    listener)
            .setBulkActions(2000)
            // 总文档体积达到10MB时提交
            .setBulkSize(new ByteSizeValue(10, ByteSizeUnit.MB))
            // 每XX ms提交一次（无论文档数量、体积是否达到阈值）
            .setFlushInterval(TimeValue.timeValueMillis(30))
            // 加1后为可并行的提交请求数，即设为0代表只可1个请求并行，设为1为2个并行
            .setConcurrentRequests(5).build();

    private static void initTransportClient() {

        String addresses = "localhost";
        Integer port = 9200;
        String[] addressArray = addresses.split(",");

        HttpHost[] httpHosts = new HttpHost[addressArray.length];
        for (int i = 0; i < addressArray.length; i++) {
            httpHosts[i] = new HttpHost(addressArray[i], port, CONNECTION_TYPE);
        }

        client = new RestHighLevelClient(RestClient.builder(httpHosts));
    }

    public static RestHighLevelClient getClient() {
        return client;
    }

    public static long getCount(String[] indices, String[] types, QueryBuilder queryBuilder) {
        if (indices == null || indices.length == 0) {
            return 0;
        }

        SearchSourceBuilder searchRequestBuilder = new SearchSourceBuilder();
        searchRequestBuilder.query(queryBuilder);
        searchRequestBuilder.from(0);
        searchRequestBuilder.size(0);
        searchRequestBuilder.explain(false);
        searchRequestBuilder.trackTotalHits(true);
        SearchRequest searchRequest = new SearchRequest(indices);
        searchRequest.source(searchRequestBuilder);
        searchRequest.searchType(SearchType.DFS_QUERY_THEN_FETCH);

        try {
            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
            return response.getHits().getTotalHits().value;
        } catch (IOException ex) {
            LOGGER.error("ElasticSearchRestManager getCount error " + ex.getMessage(), ex);
        }
        return 0;
    }

    /**
     * 查询去重后总量
     */
    public static Integer countDistinct(String[] indices, String[] types, String field) {
        // 创建查询请求对象
        SearchRequest searchRequest = new SearchRequest(indices);
        // 创建查询资源对象
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        // 查询条件-分组
        CardinalityAggregationBuilder cardinalityAggregationBuilder = AggregationBuilders.cardinality(field).field(field);
        sourceBuilder.aggregation(cardinalityAggregationBuilder);
        sourceBuilder.size(0);
        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = null;
        int size = 0;
        try {
            searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            if (null == searchResponse) {
                return 0;
            }
            ParsedCardinality parsedCardinality = searchResponse.getAggregations().get(field);
            // 这个field就是上面起的别名
            size = (int) parsedCardinality.getValue();
        } catch (IOException e) {
            LOGGER.debug("ES查询分组条数有误！");
        }
        return size;
    }


    public static Map<String, Object> findById(String index, String id) {

        GetRequest getRequest = new GetRequest();
        getRequest.id(id);
        getRequest.index(index);
        try {
            GetResponse response = client.get(getRequest, RequestOptions.DEFAULT);
            if (response.isExists()) {
                return response.getSourceAsMap();
            }
        } catch (IOException ex) {
            LOGGER.error("ElasticSearchRestManager findById error " + ex.getMessage(), ex);
        }
        return null;
    }

    public static List<Map<String, Object>> getList(String[] indices, String[] types,
                                                    QueryBuilder queryBuilder, int currentPage, int size, String sortedField) {
        if (size == 0) {
            size = ITEMS_PER_SEARCH;
        } else if (size > ES_MAX_SIZE) {
            size = ES_MAX_SIZE;
            LOGGER.warn(
                    "query index size exceed 10000.  indices ：[{}],types : 【{}】,queryBuilder : 【{}】，currentPage ：【{}】,size : 【{}】,sortedField : 【{}】",
                    new Object[]{
                            indices, types, queryBuilder, currentPage, size, sortedField});
        }
        List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
        if (indices == null || indices.length == 0) {
            return result;
        }

        SearchSourceBuilder searchRequestBuilder = new SearchSourceBuilder();
        searchRequestBuilder.query(queryBuilder);
        if (StringUtils.isNotEmpty(sortedField)) {
            searchRequestBuilder.sort(sortedField, SortOrder.DESC);
        }
        searchRequestBuilder.from(currentPage * size);
        searchRequestBuilder.size(size);
        searchRequestBuilder.explain(false);
        SearchRequest searchRequest = new SearchRequest(indices);
        searchRequest.source(searchRequestBuilder);
        searchRequest.searchType(SearchType.DFS_QUERY_THEN_FETCH);

        try {
            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = response.getHits();

            if (hits != null) {
                SearchHit[] searchHits = hits.getHits();
                for (SearchHit searchHit : searchHits) {
                    Map<String, Object> map = searchHit.getSourceAsMap();
                    map.put(ES_INDEX, searchHit.getIndex());
                    map.put(ES_TYPE, searchHit.getType());
                    result.add(formatMap(searchHit.getSourceAsMap()));
                }
            }
        } catch (IOException ex) {
            LOGGER.error(ex.getMessage(), ex);
        }
        return result;
    }

    public static PaginationBean<Map<String, Object>> getPagination(String[] indices, String[] types,
                                                                    QueryBuilder queryBuilder, int currentPage, int size, String sortedField) {
        if (size == 0) {
            size = ITEMS_PER_SEARCH;
        } else if (size > ES_MAX_SIZE) {
            size = ES_MAX_SIZE;
            LOGGER.warn(
                    "query index size exceed 10000.  indices ：[{}],types : 【{}】,queryBuilder : 【{}】，currentPage ：【{}】,size : 【{}】,sortedField : 【{}】",
                    new Object[]{
                            indices, types, queryBuilder, currentPage, size, sortedField});
        }
        List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
        PaginationBean<Map<String, Object>> paginationBean = new PaginationBean<>();
        if (indices == null || indices.length == 0) {
            return paginationBean;
        }

        SearchSourceBuilder searchRequestBuilder = new SearchSourceBuilder();
        searchRequestBuilder.query(queryBuilder);
        if (StringUtils.isNotEmpty(sortedField)) {
            searchRequestBuilder.sort(sortedField, SortOrder.DESC);
        }
        searchRequestBuilder.from(currentPage * size);
        searchRequestBuilder.size(size);
        searchRequestBuilder.explain(false);
        searchRequestBuilder.trackTotalHits(true);
        SearchRequest searchRequest = new SearchRequest(indices);
        searchRequest.source(searchRequestBuilder);
        searchRequest.searchType(SearchType.DFS_QUERY_THEN_FETCH);

        try {
            SearchResponse response = ElasticSearchRestManager.client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = response.getHits();

            if (hits != null) {
                SearchHit[] searchHits = hits.getHits();
                for (SearchHit searchHit : searchHits) {
                    Map<String, Object> map = searchHit.getSourceAsMap();
                    map.put(ES_INDEX, searchHit.getIndex());
                    result.add(formatMap(searchHit.getSourceAsMap()));
                }
                paginationBean.setTotalRows(hits.getTotalHits().value);
            }
        } catch (IOException ex) {
            LOGGER.error("ElasticsearchRestManager getPagination error " + ex.getMessage(), ex);
        }

        paginationBean.setResult(result);
        return paginationBean;
    }

    public static List<List> getDocumentList(String[] indices, String[] types,
                                             QueryBuilder queryBuilder, String[] fieldList, String orderField, SearchOrder order, int start, int size) {

        if (size == 0) {
            size = ITEMS_PER_SEARCH;
        } else if (size > ES_MAX_SIZE) {
            size = ES_MAX_SIZE;
            LOGGER.warn(
                    "query index size exceed 10000.  indices ：[{}],types : 【{}】,queryBuilder : 【{}】，start ：【{}】,size : 【{}】,orderField : 【{}】",
                    new Object[]{
                            indices, types, queryBuilder, start, size, orderField});
        }

        List<List> result = new ArrayList();
        if (indices == null || indices.length == 0) {
            return result;
        }


        SearchSourceBuilder searchRequestBuilder = new SearchSourceBuilder();
        searchRequestBuilder.query(queryBuilder);
        if (StringUtils.isNotEmpty(orderField)) {
            SortBuilder sortBuilder = getSortBuilder(orderField, order);
            searchRequestBuilder.sort(sortBuilder);
        }
        searchRequestBuilder.from(start);
        searchRequestBuilder.size(size);
        SearchRequest searchRequest = new SearchRequest(indices);
        searchRequest.source(searchRequestBuilder);
        try {
            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = response.getHits();
            SearchHit[] searchHits = hits.getHits();

            for (SearchHit hit : searchHits) {
                if (!ArrayUtils.isEmpty(fieldList)) {
                    Map<String, Object> map = hit.getSourceAsMap();
                    List<Object> theField = new ArrayList();
                    for (String field : fieldList) {
                        if (map.get(field) == null) {
                            theField.add("");
                            continue;
                        }
                        theField.add(map.get(field));
                    }
                    result.add(theField);
                }
            }
        } catch (Exception e) {
            LOGGER.info("error query index!", e);
        }
        return result;
    }


    public static List<Object> distinct(String index, String field, QueryBuilder queryBuilder) {


        SearchSourceBuilder builder = new SearchSourceBuilder()
                .query(queryBuilder)
                .collapse(new CollapseBuilder(field + ".keyword"))
                .fetchSource(new String[]{field}, null);

        SearchRequest request = new SearchRequest(index);
        request.source(builder);

        List<Object> list = new ArrayList<>();
        try {
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            for (SearchHit hit : response.getHits().getHits()) {
                list.add(hit.getSourceAsMap().get(field));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        return list;
    }


    public static boolean updateByQueryMap(String index, String type, String id, Map<String, Object> map) {
        boolean result = false;
        try {
            XContentBuilder source = XContentFactory.jsonBuilder();
            source.map(map);

            UpdateRequest request = new UpdateRequest(index, id).doc(source).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).routing(id);

            UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
            if (RestStatus.OK.equals(updateResponse.status())) {
                result = true;
            }
        } catch (IOException | DocumentMissingException e) {
            LOGGER.info("error update index!", e);
        }
        return result;
    }

    public static boolean delete(Client elasticSearchClient, String id, String index, String type) {
        return elasticSearchClient.prepareDelete(index, type, id)
                .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
                .execute()
                .isDone();
    }

    public static boolean multiDelete(BulkProcessor staticBulkProcessor, String index, String type, String... ids) {
        DeleteRequest deleteRequest;
        for (String id : ids) {
            deleteRequest = new DeleteRequest(index, type, id);
            staticBulkProcessor.add(deleteRequest);
        }
        return true;
    }

    public static boolean insertObject(String index, String id, Object data) {

        try {
            IndexRequest indexRequest = new IndexRequest(index);
            indexRequest.id(id);
            indexRequest.source(JSONObject.toJSONString(data), XContentType.JSON);
            IndexResponse response = client.index(indexRequest, RequestOptions.DEFAULT);
            LOGGER.info("增加记录:{},状态码:{}",JSONObject.toJSONString(response),response.status().getStatus());
        } catch (Exception e) {
            LOGGER.debug("添加文档失败",e);
            return false;
        }
        return true;
    }

    public static boolean insert(String index, String id, Map<String, ?> map) {
        boolean isUpdateSuccess = true;
        // 空值看作更新成功
        if (map == null) {
            return true;
        }

        try {
            IndexRequest request = new IndexRequest(index).id(id);
            request.source(map, XContentType.JSON);
            request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
            client.index(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            LOGGER.error("ElasticSearchRestManager insert error " + e.getMessage(), e);
            isUpdateSuccess = false;
        } catch (RuntimeException e) {
            LOGGER.error("error delete index by ID!", e);
            isUpdateSuccess = false;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return isUpdateSuccess;
    }

    public static boolean insertByBulk(String index, String id, Map map) {
        boolean result = true;
        try {
            // EsClient.getBulkProcessor()是位于上方EsClient类中的方法
            bulkProcessor.add(new IndexRequest(index, "_doc", id).source(map));//添加文档，以便自动提交
        } catch (Exception e) {
            result = false;
            LOGGER.info("add文档时出现异常：e=" + e + " map=" + map, e);
        }
        return result;
    }

    public static boolean updateByBulk(String index, String id, Map<String, Object> map) {
        try {
            bulkProcessor.add(new UpdateRequest(index, "_doc", id).doc(map));
        } catch (Exception e) {
            LOGGER.info("error update index!", e);
        }
        return true;
    }

    private static Map<String, Object> formatMap(Map<String, Object> elasticSearchMap) {
        elasticSearchMap.remove("_version_");
        return elasticSearchMap;
    }

    private static SortBuilder getSortBuilder(String orderField, SearchOrder order) {
        SortBuilder sortBuilder = null;
        if (StringUtils.isNotBlank(orderField)) {
            if (order != null) {
                switch (order) {
                    case ASC:
                        sortBuilder = SortBuilders.fieldSort(orderField).unmappedType("long").order(SortOrder.ASC);
                        break;

                    case DESC:
                        sortBuilder = SortBuilders.fieldSort(orderField).unmappedType("long").order(SortOrder.DESC);
                        break;

                    default:
                        LOGGER.error("order=[{}]", order);
                        break;
                }
            } else {
                sortBuilder = SortBuilders.fieldSort(orderField).unmappedType("long").order(SortOrder.DESC);
            }
        }

        return sortBuilder;
    }

    public static String[] getExitIndices(String[] indices) {
        if (null == indices) {
            return new String[]{};
        }
        if (indices.length < 1) {
            return indices;
        }
        List<String> list = new ArrayList<>();
        try {

//解决长度过长问题
//            GetIndexRequest getIndexRequest = new GetIndexRequest(indices);
//
//            boolean indicesExist =
//                    elasticSearchClient.indices().exists(getIndexRequest, RequestOptions.DEFAULT);
//            if (indicesExist) {
//                return indices;
//            } else {
            boolean indicesExist;
            for (String str : indices) {
                indicesExist = client.indices().exists(new GetIndexRequest(str), RequestOptions.DEFAULT);
                if (indicesExist) {
                    list.add(str);
                }
            }
//            }
        } catch (IOException ex) {

        }
        return list.toArray(new String[list.size()]);
    }


    public static boolean isExistsIndex(String index) {

        try {
            boolean responseExist =
                    getClient().indices().exists(new GetIndexRequest(index), RequestOptions.DEFAULT);
            return responseExist;
        } catch (IOException ex) {
            LOGGER.error("ElasticSearchRestManager isExistsIndex error" + ex.getMessage(), ex);
        }
        return false;
    }

    public static String[] createIndices(final String... indices) {

        try {
            if (null == indices || indices.length < 1) {
                return indices;
            }

            boolean responseExist = getClient().indices().exists(new GetIndexRequest(indices), RequestOptions.DEFAULT);

            if (responseExist) {
                return indices;
            } else {
                for (String str : indices) {
                    boolean response = isExistsIndex(str);
                    if (!response) {
                        try {
                            getClient().indices().exists(new GetIndexRequest(str), RequestOptions.DEFAULT);
                        } catch (ResourceAlreadyExistsException e) {
                            LOGGER.error("索引已经存在！");
                        }

                    }
                }
            }
            return indices;
        } catch (IOException ex) {
            LOGGER.error("createIndices error " + ex.getMessage(), ex);
            return null;
        }
    }

}
